package com.ptkc.utils;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @author: Mr.Jiao
 * @date: 2018年12月20日
 */
public abstract class DelayQueueRunner<V> extends Thread implements Runnable {

    private final DelayQueue<Element> delayQueue = new DelayQueue<>();

    private final ConcurrentHashMap<String, V> concurrentHashMap = new ConcurrentHashMap();

    private class Element implements Delayed {

        private long start = System.currentTimeMillis();

        private String value;

        private long delayTime;

        public Element(String value, long delayTime) {
            this.value = value;
            this.delayTime = delayTime;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert((start + delayTime) - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }

    }

    @Override
    public final void run() {
        while (true){
            Element element = null;
            try {
                element = delayQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if(element != null && concurrentHashMap.containsKey(element.value)) {
                V v = concurrentHashMap.remove(element.value);
                this.run(element.value, v);
            }
        }
    }

    public abstract void run(String k, V v);


    public final void put(String k, V v, long delay){
        if (!concurrentHashMap.containsKey(k)) {
            concurrentHashMap.put(k, v);
            delayQueue.offer(new Element(k, delay));
        }
    }

    @Override
    public final synchronized void start() {
        if (this.getState() == State.NEW){
            super.start();
        }else{
            throw new RuntimeException("线程已经在运行");
        }
    }

}
